package com.intct.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.concurrent.ExecutionException;

/**
 * @author gufg
 * @since 2025-07-01 11:38
 */
public class FlinkSQLDemo4 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 创建临时，连接mysql
        tEnv.executeSql("create table flink_sql(" +
                "  id INTEGER," +
                "  name STRING," +
                "  age INTEGER," +
                "  sex char(10)," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "   'connector' = 'jdbc'," +
                "   'url' = 'jdbc:mysql://jd-node:3306/m1'," +
                "   'table-name' = 'flink_sql', 'driver' = 'com.mysql.cj.jdbc.Driver', 'username' = 'root', 'password' = '999999'" +
                ")");

        // select * from flink_sql
        Table table = tEnv.sqlQuery("select * from flink_sql");

        // kafa sink
        // 创建临时表，连接Kafka
        tEnv.executeSql("CREATE TABLE pageviews_per_region (" +
                "  id INTEGER," +
                "  name STRING," +
                "  age INTEGER," +
                "  sex char(10)," +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'upsert-kafka'," +
                "  'topic' = 'intct81'," +
                "  'properties.bootstrap.servers' = 'cdh-node:9092'," +
                "  'key.format' = 'json'," +
                "  'value.format' = 'json'" +
                ")");

        // 写入数据
//        table.executeInsert("pageviews_per_region");
        tEnv.executeSql("insert into pageviews_per_region select * from flink_sql");
    }
}
